package HandshakeJoin;

import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.io.FileReader;
import java.io.IOException;
import java.io.FileNotFoundException;
import com.google.common.collect.Lists;
import com.google.common.io.LineReader;
import com.google.common.base.Splitter;
import org.apache.s4.core.ProcessingElement;
import org.apache.s4.core.RemoteStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SourceConverter extends ProcessingElement
{
    transient private static final Logger logger = LoggerFactory
            .getLogger(SourceConverter.class);

    private RemoteStream outputStream;
    private int sourceId = 0;
    private String sourceFilename = null;
    private int numFields = 4;
    private int sizeOfOneEmit = 8;
    private Random random;
    LineReader lineReader = null;

    @Override
    protected void onCreate() {
        random = new Random(outputStream.getName().hashCode());

        if (sourceFilename != null) {
            try {
                FileReader reader = new FileReader(sourceFilename);
                lineReader = new LineReader(reader);
            }
            catch (FileNotFoundException e) {
                logger.warn("file not found: " + sourceFilename);
            }
        }
    }

    public void onEvent(DataEvent data) {
        outputStream.put(data);
    }

    public void onTime() {
        emitLinesOfSourceFile();
    }

    @Override
    protected void onRemove() {}

    private void emitLinesOfSourceFile() {
        if (lineReader == null) {
            return;
        }

        for (int i = 0; i < sizeOfOneEmit; ++i) {
            String line = null;
            try {
                line = lineReader.readLine();
            }
            catch (IOException e) {
                e.printStackTrace();
            }

            if (line != null) {
                DataEvent data = constructDataEvent(line);
                outputStream.put(data);
            }
            else {
                break;
            }
        }
    }

    private DataEvent constructDataEvent(final String line) {
        DataEvent data = new DataEvent(sourceId, numFields);
        Iterable<String> split = Splitter.on(',').omitEmptyStrings()
                .trimResults().split(line);

        int i = 0;
        for (String term : split) {
            data.setField(i++, Integer.parseInt(term));
        }

        return data;
    }

    public SourceConverter setOutputStream(final RemoteStream outputStream) {
        this.outputStream = outputStream;

        return this;
    }

    public SourceConverter setSourceId(final int id) {
        this.sourceId = id;

        return this;
    }

    public SourceConverter setSourceFilename(final String filename) {
        this.sourceFilename = filename;

        return this;
    }

    public SourceConverter setEmitInterval(final long interval) {
        setTimerInterval(interval, TimeUnit.MILLISECONDS);

        return this;
    }

    public SourceConverter setSizeOfOneEmit(final int size) {
        this.sizeOfOneEmit = size;

        return this;
    }

    public SourceConverter setNumberOfFields(final int numFields) {
        this.numFields = numFields;

        return this;
    }
}